package com.ds.infra.mq.core.network;

import com.ds.infra.mq.core.config.BrokerConfig;
import com.ds.infra.mq.core.config.NettyConfig;
import com.ds.infra.mq.core.protocol.ProtocolCodec;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;

/**
 * 作用：Netty服务端
 *
 * @author WeiShaoying
 */
@Slf4j
public class NettyServer {

    private final BrokerConfig brokerConfig;
    private final NettyConfig nettyConfig;
    private final ServerHandler serverHandler;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    public NettyServer(BrokerConfig brokerConfig, NettyConfig nettyConfig, ServerHandler serverHandler) {
        this.brokerConfig = brokerConfig;
        this.nettyConfig = nettyConfig;
        this.serverHandler = serverHandler;
    }

    public void start() throws InterruptedException {
        bossGroup = new NioEventLoopGroup(brokerConfig.getIoThreads());
        workerGroup = new NioEventLoopGroup(brokerConfig.getWorkerThreads());

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, nettyConfig.getSoBacklog())
                .childOption(ChannelOption.TCP_NODELAY, nettyConfig.isTcpNoDelay())
                .childOption(ChannelOption.SO_KEEPALIVE, nettyConfig.isSoKeepAlive())
                .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, nettyConfig.getWriteBufferHighWaterMark())
                .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, nettyConfig.getWriteBufferLowWaterMark())
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                                .addLast(new ProtocolCodec.Decoder())
                                .addLast(new ProtocolCodec.Encoder())
                                .addLast(serverHandler);
                    }
                });

        Channel channel = bootstrap.bind(brokerConfig.getPort()).sync().channel();
        log.info("Netty server started on port {}", brokerConfig.getPort());
        channel.closeFuture().sync();
    }

    public void shutdown() {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
        log.info("Netty server shutdown");
    }
}
